package com.xunk.equipment.listener;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.xunk.common.core.constant.CacheConstants;
import com.xunk.common.core.constant.HttpStatus;
import com.xunk.common.core.utils.SpringUtils;
import com.xunk.common.core.utils.StringUtils;
import com.xunk.common.redis.service.RedisService;
import com.xunk.equipment.annotation.MqttListener;
import com.xunk.equipment.domain.EquipmentDO;
import com.xunk.equipment.mapper.EquipmentMapper;
import com.xunk.equipment.service.acrel.constant.AcrelControlAckTypeConstant;
import com.xunk.equipment.service.acrel.constant.AcrelEquipmentTypeEnum;
import com.xunk.equipment.service.acrel.constant.AcrelMqttMessageTypeConstant;
import com.xunk.equipment.service.acrel.service.AcrelSmartService;
import com.xunk.equipment.service.acrel.vo.AcrelFlowmeterDetailVO;
import com.xunk.equipment.service.acrel.vo.AcrelMqttMessage;
import com.xunk.equipment.websocket.WebsocketManager;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @ProjectName: xunk
 * @Package: com.xunk.equipment.listener.acrel
 * @ClassName: AcrelListener
 * @Author: dongxiaoxin
 * @Description: 安科瑞订阅topic 业务处理
 * @Date: 2021/3/29 19:02
 * @Version: 1.0
 */
@Slf4j
@Service
@MqttListener(topic = "v1/smartec/heatmeter/+", type = 1)
public class AcrelListener implements IMqttMessageListener {

    @Resource(name = "asyncMqttTaskExecutor")
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    RedisService redisService;

    @Autowired
    EquipmentMapper equipmentMapper;

    @Override
    public void messageArrived(String topic, MqttMessage message) {

        threadPoolTaskExecutor.execute(() -> {
            AcrelMqttMessage<?> response = JSONObject.parseObject(new String(message.getPayload()),
                    new TypeReference<AcrelMqttMessage<?>>() {
                    });
//            log.info("=========== {}", JSON.toJSONString(response));
            if (StringUtils.isNotEmpty(response.getType())) {
                String gateId = response.getGateId();
                AcrelSmartService service;
                switch (response.getType()) {
                    case AcrelMqttMessageTypeConstant.REPORT_TYPE:
                        //  数据上报
                        for (AcrelFlowmeterDetailVO<?> vo : response.getMeter()) {
                            service = SpringUtils.getBean(AcrelEquipmentTypeEnum.getEquipmentBeanName(vo.getName()).getBeanName());
                            service.saveInfluxData(gateId, vo, response.getTime());
                        }
                        break;
                    case AcrelMqttMessageTypeConstant.ALARM_TYPE:
                        //  警告 TODO 需要推送app
                        log.info("收到警告了！！！");
                        log.info(new String(message.getPayload()));
                        for (AcrelFlowmeterDetailVO<?> vo : response.getMeter()) {
                            service = SpringUtils.getBean(AcrelEquipmentTypeEnum.getEquipmentBeanName(vo.getName()).getBeanName());
                            service.saveAlarmData(gateId, vo);
                        }
                        break;
                    case AcrelMqttMessageTypeConstant.CONTROL_ACK_TYPE:
                        //  控制反馈
                        log.info("Enter -> Control_ack");
                        log.info(new String(message.getPayload()));
                        Map<String, Object> map = redisService.getCacheObject(CacheConstants.ACREL_CONTROL_UUID + response.getCuuid());
                        String controlKey = AcrelControlAckTypeConstant.HEAT_CONTROL_FUNCTION.equals(response.getFunctionId()) ?
                                CacheConstants.HEAT_SWITCH_STATUS
                                :
                                CacheConstants.WATER_SWITCH_STATUS;
                        //  根据反馈数据 - 更新相应设备开关状态
                        redisService.setCacheObject(controlKey + map.get("id"),
                                AcrelControlAckTypeConstant.ACK_SUCCESS.equals(response.getResult()) ?
                                        map.get("status")
                                        :
                                        redisService.getCacheObject(controlKey + ":last:" + map.get("id")));
                        sendWebSocketMessage(response.getGateId() + ":" + response.getMeterId(),
                                AcrelControlAckTypeConstant.ACK_SUCCESS.equals(response.getResult()),map.get("status").toString());
                        break;
                    default:
                        //  心跳
                }
                //  检测整个安科瑞网关的在线状态
                redisService.setCacheObject(CacheConstants.ACREL_HEARTBEAT_TIME + response.getGateId(), response.getTime());
            }

        });

    }

    public void sendWebSocketMessage(String deviceSn, Boolean result,String status){

        Map<String,Object> dataMap = new HashMap<>();
        dataMap.put("on",status);

        Map<String,Object> map = new HashMap<>();
        map.put("deviceSn",deviceSn);
        EquipmentDO equipment = equipmentMapper.getEquipment(deviceSn);
        map.put("deviceType",equipment.getDeviceType());
        map.put("properties",dataMap);

        Map<String,Object> socketMessage = new HashMap<>();
        socketMessage.put("code", result ? HttpStatus.SUCCESS : HttpStatus.ERROR);
        socketMessage.put("msg", "设备执行"+(result ? "成功" : "失败"));
        socketMessage.put("type", 3);

        socketMessage.put("data", map);
        List<Long> users = equipmentMapper.selectUsersByDeviceSn(deviceSn);
        users.forEach(user -> WebsocketManager.appointSending(String.valueOf(user),JSONObject.toJSONString(socketMessage)));
    }
}
